package com.wdl.networkflow.pv

import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.util.Collector

/**
 * 自定义 ProcessFunction，将聚合结果按窗口合并
 */
class TotalPvCountResult() extends KeyedProcessFunction[Long, PvCountView, PvCountView] {

  /** 定义一个状态，用来保存当前所有结果之和 */
  lazy val totalCountState: ValueState[Long] =
    getRuntimeContext.getState(new ValueStateDescriptor[Long]("total-count", classOf[Long]))

  override def processElement(value: PvCountView,
                              ctx: KeyedProcessFunction[Long, PvCountView, PvCountView]#Context,
                              out: Collector[PvCountView]): Unit ={
    /** 加上新的 count 值，更新状态 */
    totalCountState.update(totalCountState.value() + value.count)

    /** 注册定时器 */
    ctx.timerService().registerEventTimeTimer(value.windowEnd + 1)
  }

  override def onTimer(timestamp: Long,
                       ctx: KeyedProcessFunction[Long, PvCountView, PvCountView]#OnTimerContext,
                       out: Collector[PvCountView]): Unit = {
    /** 定时器触发时，所有分区 count 值都已经到大，输出总和 */
    out.collect(PvCountView(ctx.getCurrentKey, totalCountState.value()))
    totalCountState.clear()
  }
}
